package com.owen.flink.batch.scala.sql

import org.apache.flink.api.java.io.jdbc.JDBCInputFormat
import org.apache.flink.api.java.typeutils.RowTypeInfo
import org.apache.flink.api.scala.ExecutionEnvironment
import org.apache.flink.api.scala.typeutils.Types
import org.apache.flink.api.scala._

object FlinkSQL {

  def main(args: Array[String]) {
    val env = ExecutionEnvironment.getExecutionEnvironment

    val rowTypeInfo = new RowTypeInfo(Types.INT, Types.STRING)

    val selectQuery = "select * from test"
    val driverName = "com.mysql.cj.jdbc.Driver"
    val dbURL = "jdbc:mysql://192.168.1.254:3306/system_set"
    val dbPassword = "root"
    val dbUser = "root"

    val inputBuilder = JDBCInputFormat.buildJDBCInputFormat()
      .setDrivername(driverName)
      .setDBUrl(dbURL)
      .setQuery(selectQuery)
      .setRowTypeInfo(rowTypeInfo)
      .setUsername(dbUser)
      .setPassword(dbPassword)

    //Get Data from SQL Table
    env.createInput(inputBuilder.finish()).map(x => {
      (x.toString, 1)
    }).print()

  }


}
